package com.atguigu.flink.chapter07.state;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Arrays;

public class Flink01_Operator_ListState {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        //3s一个快照
        env.enableCheckpointing(3000);

        SingleOutputStreamOperator<String> straem = env
                .socketTextStream("hadoop162", 9999)
                .flatMap(new MyFlatMapFunction());
        straem.print();
        straem.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                if (value.contains("x")) {
                    throw new RuntimeException("故意设置一个异常，让flink程序自动重启");
                }
            }
        })
        .print();
        env.execute();
    }

    public static class MyFlatMapFunction implements FlatMapFunction<String,String>, CheckpointedFunction {

        ArrayList<String> list = new ArrayList<>();
        private ListState<String> wotdState;

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {

            String[] words = value.split(" ");
           /* for (String word : words) {
                list.add(word)
            }*/
            list.addAll(Arrays.asList(words));
            out.collect(list.toString());

        }
        // 周期性的把状态进行快照（保存）
        // 每个周期，每个 并行度  执行一次
        @Override
        public void snapshotState(FunctionSnapshotContext ctx) throws Exception {
            System.out.println("周期快照");

            //把要保存的数据，存储到状态中
           /* for (String word : list) {
                wotdState.add(word);
            }*/
            // 更新状态
            wotdState.update(list);


        }
        //程序启动，重启的时候，自动从保存的状态中恢复数据
        //一个并行度 执行一次
        @Override
        public void initializeState(FunctionInitializationContext ctx) throws Exception {
            System.out.println("重启，启动初始化");

            // 获取一个列表状态
            wotdState = ctx
                    .getOperatorStateStore()
                    .getListState(new ListStateDescriptor<String>("wordState", String.class));

            //把状态中的而数据恢复到list 集合中
            Iterable<String> words = wotdState.get();
            for (String word : words) {
                list.add(word);
            }


            //for (String word : words) list.add(word);



        }


    }


}
